package com.atguigu.flink.watermark;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.utils.MyUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/2/1
 *
 *
 *     水印的应用:
 *          ①基于eventtime的窗口运算
 *          ②定时器
 *          ③CEP
 *
 *     窗口是窗口，水印是水印，没有任何关系！
 *      窗口：本质是一个集合(bucket 桶)，用来存储数据。
 *              数据落入哪个窗口，看数据的eventtime 是否符合窗口的范围！
 *
 *      水印： 本质是EventTime计算场景下的 时间。用来更新算子的时钟，算子可以知道是否可以执行运输！
 */
public class Demo2_WaterMarkApply
{
    public static void main(String[] args) {

       Configuration conf = new Configuration();
       conf.setInteger("rest.port", 3333);
       StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.getConfig().setAutoWatermarkInterval(2000);

        //创建水印生成策略
        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.getTs());

        env.setParallelism(1);


        env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy)
            .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
            /*
                    size = slide = 5s
                    [0,4999] ,[5000,9999]
             */
            .process(new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>()
            {
                @Override
                public void process(Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                    System.out.println(context.window());
                    out.collect(MyUtil.parseList(elements).toString());
                }
            }).startNewChain()
            .print();


                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
